//
//package com.alibaba.datax.plugin.writer.influxdbwriter;
//
//import com.alibaba.datax.common.element.Column;
//import com.alibaba.datax.common.element.Record;
//import com.alibaba.datax.common.exception.DataXException;
//import com.alibaba.datax.common.plugin.RecordReceiver;
//import com.alibaba.datax.common.spi.Writer;
//import com.alibaba.datax.common.util.Configuration;
//
//import java.util.ArrayList;
//import java.util.Arrays;
//import java.util.HashMap;
//import java.util.Iterator;
//import java.util.List;
//import java.util.Map;
//import java.util.concurrent.TimeUnit;
//
//import org.apache.commons.lang3.StringUtils;
//import org.influxdb.InfluxDB;
//import org.influxdb.InfluxDBFactory;
//import org.influxdb.dto.Point;
//import org.influxdb.dto.Query;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//public class InfluxDBWriter extends Writer {
//    private static SourceDBType DB_TYPE;
//    private static InfluxDB influxdb = null;
//    private static int batchSize;
//    private static String strTimeUnit = "s";
//    private static TimeUnit timeUnit;
//    static final String TAG = "tag";
//    static final String NUM_FIELD = "number_field";
//    static final String STRING_FIELD = "string_field";
//    static final String FIELD = "field";
//    static final String TIMESTAMP = "timestamp";
//
//    public static void setEnableBatch() {
//        if (batchSize > 0 && !influxdb.isBatchEnabled()) {
//            influxdb.enableBatch(batchSize, 100, TimeUnit.MILLISECONDS);
//        }
//
//    }
//
//    static {
//        timeUnit = TimeUnit.SECONDS;
//    }
//
//    public static class Task extends Writer.Task {
//        private static final Logger LOG = LoggerFactory.getLogger(Task.class);
//        private List<String> columnName;
//        private List<String> columnType;
//        private String measurement;
//        private String database;
//        private boolean createDatabase;
//        private String retentionPolicy;
//        private int maxRetryTime;
//        public void init() {
//            Configuration writerSliceConfig = this.getPluginJobConf();
//            this.database = writerSliceConfig.getString("database");
//            this.createDatabase = writerSliceConfig.getBool("createDatabase", false);
//            this.retentionPolicy = writerSliceConfig.getString("retentionPolicy", "autogen");
//
//            try {
//                if (this.createDatabase) {
//                    InfluxDBWriter.influxdb.query(new Query("CREATE DATABASE " + this.database));
//                }
//
//                InfluxDBWriter.influxdb.setDatabase(this.database);
//                InfluxDBWriter.influxdb.setRetentionPolicy(this.retentionPolicy);
//            } catch (Exception var3) {
//                LOG.error("CREATE DATABASE " + this.database + " failed, please confirm that the Influxdb user has administrator privileges!");
//                throw DataXException.asDataXException(InfluxDBWriterErrorCode.RUNTIME_EXCEPTION, var3);
//            }
//
//            if (InfluxDBWriter.DB_TYPE == SourceDBType.RDB) {
//                this.columnName = writerSliceConfig.getList("column", String.class);
//                this.columnType = writerSliceConfig.getList("columnType", String.class);
//                if (this.columnName.size() < 3) {
//                    throw DataXException.asDataXException(InfluxDBWriterErrorCode.ILLEGAL_VALUE, "The parameter [column] is valid, which should have more than 3 elements.");
//                }
//
//                if (this.columnType.size() < 3) {
//                    throw DataXException.asDataXException(InfluxDBWriterErrorCode.ILLEGAL_VALUE, "The parameter [columnType] is valid, which should have more than 3 elements.");
//                }
//
//                if (!this.columnType.contains("tag") || !this.columnType.contains("timestamp")) {
//                    throw DataXException.asDataXException(InfluxDBWriterErrorCode.ILLEGAL_VALUE, "The parameter [columnType] is valid, which should have tag and timestamp .");
//                }
//
//                if (!this.columnType.contains("number_field") && !this.columnType.contains("string_field") && !this.columnType.contains("field")) {
//                    throw DataXException.asDataXException(InfluxDBWriterErrorCode.ILLEGAL_VALUE, "The parameter [columnType] is valid, which should have number_field or string_field or field .");
//                }
//
//                this.measurement = writerSliceConfig.getString("measurement");
//            }
//
//            this.maxRetryTime = writerSliceConfig.getInt("maxRetryTime", 3);
//        }
//
//        public void startWrite(RecordReceiver recordReceiver) {
//            long total = 0L;
//
//            try {
//                List<Record> writerBuffer = new ArrayList(InfluxDBWriter.batchSize);
//
//                while(true) {
//                    Record record;
//                    if ((record = recordReceiver.getFromReader()) == null) {
//                        if (!writerBuffer.isEmpty()) {
//                            total += this.batchWriter(writerBuffer);
//                            writerBuffer.clear();
//                        }
//                        break;
//                    }
//
//                    writerBuffer.add(record);
//                    if (writerBuffer.size() >= InfluxDBWriter.batchSize) {
//                        total += this.batchWriter(writerBuffer);
//                        writerBuffer.clear();
//                    }
//                }
//            } catch (Exception var6) {
//                LOG.error("Write fail: " + var6.getMessage());
//                throw DataXException.asDataXException(InfluxDBWriterErrorCode.RUNTIME_EXCEPTION, var6);
//            }
//
//            this.getTaskPluginCollector().collectMessage("write size ", total + "");
//            LOG.info("Task finished, write size: {}", total);
//        }
//
//        private long batchWriter(List<Record> writerBuffer) {
//            InfluxDBWriter.setEnableBatch();
//            long total = 0L;
//            if (InfluxDBWriter.DB_TYPE == SourceDBType.RDB) {
//                total = this.batchWriterRDB(writerBuffer);
//            } else if (InfluxDBWriter.DB_TYPE == SourceDBType.TSDB) {
//                total = this.batchWriterTSDB(writerBuffer);
//            } else if (InfluxDBWriter.DB_TYPE == SourceDBType.INFLUXDB) {
//                total = this.batchWriterInfluxDB(writerBuffer);
//            }
//
//            writerBuffer.clear();
//            return total;
//        }
//
//        private long batchWriterRDB(List<Record> writerBuffer) {
//            boolean sendSuc = false;
//            int fail = 0;
//            int suc = 0;
//
//            HashMap fields;
//            String name;
//            while(suc < this.maxRetryTime) {
//                try {
//                    Iterator var5 = writerBuffer.iterator();
//
//                    while(var5.hasNext()) {
//                        Record record = (Record)var5.next();
//                        fields = new HashMap();
//                        Map<String, Object> fields = new HashMap();
//                        List<Long> times = new ArrayList();
//
//                        int j;
//                        for(j = 0; j < this.columnType.size(); ++j) {
//                            name = (String)this.columnType.get(j);
//                            String name = (String)this.columnName.get(j);
//                            Column column = record.getColumn(j);
//                            if ("tag".equals(name)) {
//                                fields.put(name, column.asString());
//                            } else if ("field".equals(name)) {
//                                fields.put(name, column.asDouble());
//                            } else if ("number_field".equals(name)) {
//                                fields.put(name, column.asDouble());
//                            } else if ("string_field".equals(name)) {
//                                fields.put(name, column.asString());
//                            } else if ("timestamp".equals(name)) {
//                                times.add(column.asLong());
//                            }
//                        }
//
//                        for(j = 0; j < times.size(); ++j) {
//                            Point point = Point.measurement(this.measurement).tag(fields).time((Long)times.get(j), InfluxDBWriter.timeUnit).fields(fields).build();
//                            InfluxDBWriter.influxdb.write(point);
//                        }
//                    }
//
//                    InfluxDBWriter.influxdb.flush();
//                    sendSuc = true;
//                    break;
//                } catch (Exception var16) {
//                    ++suc;
//                    LOG.error("Write fail, retry " + suc);
//                }
//            }
//
//            if (!sendSuc) {
//                try {
//                    InfluxDBWriter.influxdb.disableBatch();
//                    Iterator var17 = writerBuffer.iterator();
//
//                    while(var17.hasNext()) {
//                        Record record = (Record)var17.next();
//                        Map<String, String> tags = new HashMap();
//                        fields = new HashMap();
//                        List<Long> times = new ArrayList();
//
//                        int j;
//                        for(j = 0; j < this.columnType.size(); ++j) {
//                            String type = (String)this.columnType.get(j);
//                            name = (String)this.columnName.get(j);
//                            Column column = record.getColumn(j);
//                            if ("tag".equals(type)) {
//                                tags.put(name, column.asString());
//                            } else if ("field".equals(type)) {
//                                fields.put(name, column.asDouble());
//                            } else if ("number_field".equals(type)) {
//                                fields.put(name, column.asDouble());
//                            } else if ("string_field".equals(type)) {
//                                fields.put(name, column.asString());
//                            } else if ("timestamp".equals(type)) {
//                                times.add(column.asLong());
//                            }
//                        }
//
//                        try {
//                            for(j = 0; j < times.size(); ++j) {
//                                Point point = Point.measurement(this.measurement).tag(tags).time((Long)times.get(j), InfluxDBWriter.timeUnit).fields(fields).build();
//                                InfluxDBWriter.influxdb.write(point);
//                            }
//                        } catch (Exception var14) {
//                            ++fail;
//                            this.getTaskPluginCollector().collectDirtyRecord(record, "Write data points failed! " + var14.getMessage());
//                        }
//                    }
//                } catch (Exception var15) {
//                    LOG.error("Write fail: " + var15.getMessage());
//                }
//            }
//
//            suc = writerBuffer.size() - fail;
//            return (long)suc;
//        }
//
//        private long batchWriterInfluxDB(List<Record> writerBuffer) {
//            boolean sendSuc = false;
//            int fail = 0;
//            int suc = 0;
//
//            int j;
//            while(suc < this.maxRetryTime) {
//                try {
//                    Iterator var5 = writerBuffer.iterator();
//
//                    while(var5.hasNext()) {
//                        Record record = (Record)var5.next();
//                        j = record.getColumnNumber();
//
//                        for(int j = 0; j < j; ++j) {
//                            DataPoint4InfluxDB dataPoint4InfluxDB = (DataPoint4InfluxDB)JSONObject.parseObject(record.getColumn(j).asString(), DataPoint4InfluxDB.class);
//                            long timestamp = dataPoint4InfluxDB.getTimestamp();
//                            Point point = Point.measurement(dataPoint4InfluxDB.getMetric()).tag(dataPoint4InfluxDB.getTags()).time(timestamp, InfluxDBWriter.timeUnit).fields(dataPoint4InfluxDB.getFields()).build();
//                            InfluxDBWriter.influxdb.write(point);
//                        }
//                    }
//
//                    InfluxDBWriter.influxdb.flush();
//                    sendSuc = true;
//                    break;
//                } catch (Exception var15) {
//                    ++suc;
//                    LOG.error("Write fail, retry " + suc);
//                }
//            }
//
//            if (!sendSuc) {
//                try {
//                    InfluxDBWriter.influxdb.disableBatch();
//                    Iterator var16 = writerBuffer.iterator();
//
//                    while(var16.hasNext()) {
//                        Record record = (Record)var16.next();
//                        int recordLength = record.getColumnNumber();
//
//                        for(j = 0; j < recordLength; ++j) {
//                            try {
//                                DataPoint4InfluxDB dataPoint4InfluxDB = (DataPoint4InfluxDB)JSONObject.parseObject(record.getColumn(j).asString(), DataPoint4InfluxDB.class);
//                                long timestamp = dataPoint4InfluxDB.getTimestamp();
//                                Point point = Point.measurement(dataPoint4InfluxDB.getMetric()).tag(dataPoint4InfluxDB.getTags()).time(timestamp, InfluxDBWriter.timeUnit).fields(dataPoint4InfluxDB.getFields()).build();
//                                InfluxDBWriter.influxdb.write(point);
//                            } catch (Exception var13) {
//                                ++fail;
//                                this.getTaskPluginCollector().collectDirtyRecord(record, "Write data points failed! " + var13.getMessage());
//                            }
//                        }
//                    }
//                } catch (Exception var14) {
//                    LOG.error("Write fail: " + var14.getMessage());
//                }
//            }
//
//            suc = writerBuffer.size() - fail;
//            return (long)suc;
//        }
//
//        private long batchWriterTSDB(List<Record> writerBuffer) {
//            boolean sendSuc = false;
//            int fail = 0;
//            int suc = 0;
//
//            int j;
//            while(suc < this.maxRetryTime) {
//                try {
//                    Iterator var5 = writerBuffer.iterator();
//
//                    while(var5.hasNext()) {
//                        Record record = (Record)var5.next();
//                        j = record.getColumnNumber();
//
//                        for(int j = 0; j < j; ++j) {
//                            DataPoint4TSDB dataPoint4TSDB = (DataPoint4TSDB)JSONObject.parseObject(record.getColumn(j).asString(), DataPoint4TSDB.class);
//                            Map<String, Object> fields = new HashMap();
//                            fields.put("value", dataPoint4TSDB.getValue());
//                            long timestamp = dataPoint4TSDB.getTimestamp();
//                            Point point = Point.measurement(dataPoint4TSDB.getMetric()).tag(dataPoint4TSDB.getTags()).time(timestamp, InfluxDBWriter.timeUnit).fields(fields).build();
//                            InfluxDBWriter.influxdb.write(point);
//                        }
//                    }
//
//                    InfluxDBWriter.influxdb.flush();
//                    sendSuc = true;
//                    break;
//                } catch (Exception var16) {
//                    ++suc;
//                    LOG.error("Write fail, retry " + suc);
//                }
//            }
//
//            if (!sendSuc) {
//                try {
//                    InfluxDBWriter.influxdb.disableBatch();
//                    Iterator var17 = writerBuffer.iterator();
//
//                    while(var17.hasNext()) {
//                        Record record = (Record)var17.next();
//                        int recordLength = record.getColumnNumber();
//
//                        for(j = 0; j < recordLength; ++j) {
//                            try {
//                                DataPoint4TSDB dataPoint4TSDB = (DataPoint4TSDB)JSONObject.parseObject(record.getColumn(j).asString(), DataPoint4TSDB.class);
//                                Map<String, Object> fields = new HashMap();
//                                fields.put("value", dataPoint4TSDB.getValue());
//                                long timestamp = dataPoint4TSDB.getTimestamp();
//                                Point point = Point.measurement(dataPoint4TSDB.getMetric()).tag(dataPoint4TSDB.getTags()).time(timestamp, InfluxDBWriter.timeUnit).fields(fields).build();
//                                InfluxDBWriter.influxdb.write(point);
//                            } catch (Exception var14) {
//                                ++fail;
//                                this.getTaskPluginCollector().collectDirtyRecord(record, "Write data points failed! " + var14.getMessage());
//                            }
//                        }
//                    }
//                } catch (Exception var15) {
//                    LOG.error("Write fail: " + var15.getMessage());
//                }
//            }
//
//            suc = writerBuffer.size() - fail;
//            return (long)suc;
//        }
//
//        public void post() {
//        }
//
//        public void destroy() {
//        }
//    }
//
//    public static class Job extends Writer.Job {
//        private static final Logger LOG = LoggerFactory.getLogger(Job.class);
//        private Configuration originalConfig;
//
//        public void init() {
//            this.originalConfig = super.getPluginJobConf();
//            String sourceDbType = this.originalConfig.getString("sourceDbType");
//            if (StringUtils.isBlank(sourceDbType)) {
//                sourceDbType = SourceDBType.RDB.name();
//                this.originalConfig.set("sourceDbType", sourceDbType);
//                LOG.info("The parameter [sourceDbType] will be default value: " + SourceDBType.TSDB);
//            }
//
//            try {
//                InfluxDBWriter.DB_TYPE = SourceDBType.valueOf(sourceDbType);
//            } catch (Exception var5) {
//                throw DataXException.asDataXException(InfluxDBWriterErrorCode.REQUIRED_VALUE, "The parameter [sourceDbType] is valid, which should be one of [" + Arrays.toString(SourceDBType.values()) + "].");
//            }
//
//            String userName;
//            if (InfluxDBWriter.DB_TYPE != SourceDBType.TSDB && InfluxDBWriter.DB_TYPE != SourceDBType.INFLUXDB) {
//                if (InfluxDBWriter.DB_TYPE == SourceDBType.RDB) {
//                    this.originalConfig.getNecessaryValue("columnType", InfluxDBWriterErrorCode.REQUIRED_VALUE);
//                    this.originalConfig.getNecessaryValue("column", InfluxDBWriterErrorCode.REQUIRED_VALUE);
//                } else {
//                    LOG.info("not support! db_type : " + InfluxDBWriter.DB_TYPE);
//                }
//            } else {
//                userName = this.originalConfig.getString("endpoint");
//                if (StringUtils.isBlank(userName)) {
//                    throw DataXException.asDataXException(InfluxDBWriterErrorCode.REQUIRED_VALUE, "The parameter [endpoint] is not set.");
//                }
//
//                Integer batchSize = this.originalConfig.getInt("batchSize");
//                if (batchSize == null || batchSize < 1) {
//                    this.originalConfig.set("batchSize", 100);
//                    LOG.info("The parameter [batchSize] will be default value: 100");
//                }
//            }
//
//            InfluxDBWriter.strTimeUnit = this.originalConfig.getString("timeUnit");
//            if (InfluxDBWriter.strTimeUnit != null && !InfluxDBWriter.strTimeUnit.isEmpty()) {
//                if (InfluxDBWriter.strTimeUnit.equals("s")) {
//                    InfluxDBWriter.timeUnit = TimeUnit.SECONDS;
//                } else if (InfluxDBWriter.strTimeUnit.equals("ms")) {
//                    InfluxDBWriter.timeUnit = TimeUnit.MILLISECONDS;
//                } else if (InfluxDBWriter.strTimeUnit.equals("us")) {
//                    InfluxDBWriter.timeUnit = TimeUnit.MICROSECONDS;
//                } else {
//                    if (!InfluxDBWriter.strTimeUnit.equals("ns")) {
//                        throw DataXException.asDataXException(InfluxDBWriterErrorCode.ILLEGAL_VALUE, "The parameter [timeUnit] must be one of s, ms ,us and ns.");
//                    }
//
//                    InfluxDBWriter.timeUnit = TimeUnit.NANOSECONDS;
//                }
//            }
//
//            this.originalConfig.getNecessaryValue("endpoint", InfluxDBWriterErrorCode.REQUIRED_VALUE);
//            this.originalConfig.getNecessaryValue("database", InfluxDBWriterErrorCode.REQUIRED_VALUE);
//            userName = this.originalConfig.getString("username");
//            String passWord = this.originalConfig.getString("password");
//            String endpoint = this.originalConfig.getString("endpoint");
//            if (endpoint.startsWith("http:")) {
//                InfluxDBWriter.influxdb = InfluxDBFactory.connect(endpoint, userName, passWord);
//            } else {
//                if (!endpoint.startsWith("https:")) {
//                    throw DataXException.asDataXException(InfluxDBWriterErrorCode.REQUIRED_VALUE, "The parameter [endpoint] is valid, which should be [http://IP:Port] or [https://IP:Port].   endpoint:" + endpoint);
//                }
//
//                InfluxDBWriter.influxdb = InfluxDBFactory.connect(endpoint, userName, passWord, (new HttpsUtils()).getTrustAllClient().newBuilder());
//            }
//
//            InfluxDBWriter.batchSize = this.originalConfig.getInt("batchSize", 100);
//            InfluxDBWriter.setEnableBatch();
//            LOG.info("InfluxDB config:" + this.originalConfig.toJSON());
//        }
//
//        public void prepare() {
//        }
//
//        public List<Configuration> split(int mandatoryNumber) {
//            ArrayList<Configuration> configurations = new ArrayList(mandatoryNumber);
//
//            for(int i = 0; i < mandatoryNumber; ++i) {
//                configurations.add(this.originalConfig.clone());
//            }
//
//            return configurations;
//        }
//
//        public void post() {
//        }
//
//        public void destroy() {
//            if (InfluxDBWriter.influxdb != null) {
//                InfluxDBWriter.influxdb.close();
//            }
//
//        }
//    }
//}
